package com.zuikaku.demo;

import com.zuikaku.pojo.Order;
import com.zuikaku.sink.OrderSink;
import com.zuikaku.source.OrderSource;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class _3SourceSinkDemo {

    public static void main(String[] args) {
        //Flink流处理案例
        //1.取得流运行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
        env.setParallelism(1);
        //来自于自定义流
        DataStream<Order> source5 = env.addSource(new OrderSource());
        source5.print("源数据");
        source5.print("iphone订单source").setParallelism(4);
        //添加自定义Sink进行消费
        source5.addSink(new OrderSink());

        try {
            env.execute("source api demo");
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}
